package hdfsClientDemo;

import com.google.common.base.Charsets;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Before;
import org.junit.Test;

import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;


/**
 * Created by LiuChang on 2017/9/29/029.
 * hdfs的客户端API功能无非就是去操作hdfs上的文件或文件夹
 */
public class HdfsClientDemo {
    FileSystem hdfsClient;

    @Before
    public void init() throws URISyntaxException, IOException, InterruptedException {
        //第一步：创建一个hdfs的客户端对象
        Configuration configuration = new Configuration();
        configuration.addResource("demo.xml");
        configuration.set("dfs.replication", "2");
        configuration.set("dfs.blocksize", "32m");
        hdfsClient = FileSystem.get(new URI("hdfs://hdp1:9000/"), configuration, "root");
    }

    @Test
    public void testUpload() throws IOException {
        //测试：上传一个文件
        hdfsClient.copyFromLocalFile(false, true, new Path("D:/jmeter/apache-jmeter-3.2.zip"), new Path("/jmeter.gz"));
        hdfsClient.close();
    }

    @Test
    public void testDownload() throws IOException {
        //hdfs客户端从hdfs上读取数据写入本地磁盘时，可以选择使用hadoop自己开发的本地库来操作，也可以使用java的元神库来操作本地文件
        //参数就是：useRawLocalFileSystem,如果为true，则使用java原生库，否则，使用hadoop自己的本地库

        //如果要是用hadoop自己的本地库来操作本地磁盘文件，需要在本地系统中配置一下hadoop的本地库目录到环境变量中
        hdfsClient.copyToLocalFile(false, new Path("/jmeter.gz"), new Path("C:/Users/Administrator/Desktop/testDownload.zip"), true);
        hdfsClient.close();
    }


    @Test
    public void testMkdir() throws IOException {
        //如果创建文件夹时，指定权限信息，当创建的结果不会跟指定的信息完全一致，因为指定的信息还会经过一个参数值的掩码运算
        //fs.permissions.umask-mode，默认值是022，因此，在默认情况下会覆盖掉组合其他人的w权限
        hdfsClient.mkdirs(new Path("/xxx/yyyy"));
        hdfsClient.close();
    }

    @Test
    public void testDeldir() throws IOException {
        hdfsClient.delete(new Path("/xxx"), true);
        hdfsClient.close();
    }

    @Test
    public void testRenamedir() throws IOException {
        hdfsClient.rename(new Path("/xxx"), new Path("/yyy"));
        hdfsClient.close();
    }

    /**
     * 递归查询指定路径下的所有文件信息
     *
     * @throws IOException the io exception
     */
    @Test
    public void listDir() throws IOException {
        RemoteIterator<LocatedFileStatus> files = hdfsClient.listFiles(new Path("/"), true);
        while (files.hasNext()) {
            LocatedFileStatus file = files.next();
            System.out.println("最近访问时间：" + file.getAccessTime());
            System.out.println("文件的分块大小：" + file.getBlockSize());
            System.out.println("文件的所属组：" + file.getGroup());
            System.out.println("文件的总长度：" + file.getLen());
            System.out.println("文件的最近修改时间：" + file.getModificationTime());
            System.out.println("文件的所有者：" + file.getOwner());
            System.out.println("文件的副本数：" + file.getReplication());
            System.out.println("文件块的位置信息：" + Arrays.toString(file.getBlockLocations()));
            System.out.println("文件的全路径：" + file.getPath());
            System.out.println("文件的权限信息：" + file.getPermission());
            System.out.println("----------------------------------------------------");

        }
        hdfsClient.close();
    }


    /**
     * 查询指定路径下的所有文件夹或者文件节点信息
     *
     * @throws IOException the io exception
     */
    @Test
    public void testListDir2() throws IOException {
        FileStatus[] listStatus = hdfsClient.listStatus(new Path("/"));
        for (FileStatus file : listStatus) {
            System.out.println(file.getPath());
            System.out.println(file.isDirectory() ? "d" : "f");
            System.out.println("----------------------------------------------------");
        }
        hdfsClient.close();
    }


    /**
     * 读取文件中指定偏移量范围的数据。从文件的0偏移量读20个字节
     *
     * @throws IOException the io exception
     */
    @Test
    public void testReadFilePart1() throws IOException {
        FSDataInputStream hdfsIn = hdfsClient.open(new Path("/qingshu.txt"));
        FileOutputStream localOut = new FileOutputStream("C:/Users/Administrator/Desktop/qingshu-part.txt");
        byte[] buf = new byte[10];
        int len = 0;
        long count = 0;
        while ((len = hdfsIn.read(buf)) != -1) {
            localOut.write(buf);
            count += len;
            if (count == 20) {
                break;
            }

        }
        localOut.flush();
        hdfsIn.close();
        localOut.close();
        hdfsClient.close();
    }


    /**
     * 读取文件中指定偏移量范围的数据。从文件的20偏移量读20个字节
     *
     * @throws IOException the io exception
     */
    @Test
    public void testReadFilePart2() throws IOException {
        FSDataInputStream hdfsIn = hdfsClient.open(new Path("/qingshu.txt"));
        //将读取的位置设置为指定的起始位置
        hdfsIn.seek(20);

        FileOutputStream localOut = new FileOutputStream("C:/Users/Administrator/Desktop/qingshu-part.txt");
        byte[] buf = new byte[10];
        int len = 0;
        long count = 0;
        while ((len = hdfsIn.read(buf)) != -1) {
            localOut.write(buf);
            count += len;
            if (count == 20) {
                break;
            }
        }
        localOut.flush();
        hdfsIn.close();
        localOut.close();
        hdfsClient.close();
    }

    @Test
    public void testWriteDataToHdfsFile() throws IOException {
        FSDataOutputStream hdfsOut = hdfsClient.create(new Path("/access.log"), true);
        hdfsOut.write("666666666666666666666".getBytes(Charsets.UTF_8));
        hdfsOut.write("888888888888888888".getBytes(Charsets.UTF_8));

        hdfsOut.flush();
        hdfsOut.close();
        hdfsClient.close();
    }

}
